package net.bwie.realtime.mall.dim.job;

import net.bwie.realtime.mall.dim.funcation.DimDataFilterFunction;
import net.bwie.realtime.mall.dim.funcation.DimHBaseMapFunction;
import net.bwie.realtime.mall.dim.sink.DimSinkHBaseFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import utils.KafkaUtil;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;

/**
 * @ClassName：RealtimeSyncDataDimJob
 * @Author: ZddddQ
 * @Date: 2025/6/3 10:03
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
public class RealtimeSyncDataDimJob {

    public static void main(String[] args) throws Exception{
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2. 数据源-source
        DataStream<String> stream = KafkaUtil.consumerKafka(env, "topic-mall");
        //stream.print("kafka");

        // 3. 数据转换-transformation
          DataStream<String> dimStream = handle(stream) ;
        // dimStream.print("dim") ;

        // 4. 数据接收器-sink
        dimStream.addSink(new DimSinkHBaseFunction());

        // 5. 触发执行-execute
        env.execute("RealtimeAsyncDataDimJob") ;
    }


    /**
     * 按照表名称过滤获取对应维表数据流，保存HBase表，列簇统一为：info，指定字段名称
     */
    private static DataStream<String> handle(DataStream<String> stream) {
       // 维度表名称与主键映射
        HashMap<String, String> dimMap = new HashMap<String, String>() {{
            put("base_dic", "dic_code"); put("base_province", "id"); put("base_region", "id");
            put("base_category1", "id"); put("base_category2", "id"); put("base_category3", "id");
            put("sku_info", "id"); put("spu_info", "id"); put("user_info", "id");
            put("activity_info", "id"); put("activity_rule", "id"); put("coupon_info", "id");
        }};
        // 维度表名称
        Set<String> dimSet = new HashSet<>(dimMap.keySet());

        // s1-过滤维表数据
        SingleOutputStreamOperator<String> syncStream = stream.filter(new DimDataFilterFunction(dimSet));
        //asyncStream.print("async");

        // s2-添加字段值
        SingleOutputStreamOperator<String> hbaseStream = syncStream.map(new DimHBaseMapFunction(dimMap));
        //hbaseStream.print("hbase");



        return hbaseStream;
    }
}
